package org.gorson.executor;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class FlinkTableDemo {

    public static void main(String[] args) {
        //TableEnvironment 可以通过静态方法 TableEnvironment.create() 创建。
/*        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                //.inBatchMode()
                .build();
        TableEnvironment.create(settings);*/

        //使用本地模式并开启WebUI
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT, "8081-8082");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

        TableEnvironment tEnv = StreamTableEnvironment.create(env);
        //1. using SQL DDL create table with jdbc connector
        tEnv.executeSql("CREATE TABLE MySqlPersonAssetTable(" +
                // declare the schema of the table 注意字段名用反引号转义
                "   trans_date STRING," +
                "   pay_code STRING," +
                "   trans_amt STRING," +
                "   trans_summary STRING," +
                "   trans_firm STRING," +
                "   trans_status STRING" +
                ") WITH(" +
                // declare the external system to connect to
                "   'connector'='jdbc'," +
                "   'url' = 'jdbc:mysql://127.0.0.1:3306/flink?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false&rewriteBatchedStatements=true'," +
                "   'table-name' = 'person_asset'," +
                "   'username' = 'root'," +
                "   'password' = '123456'" +
                ")");
        tEnv.executeSql("CREATE TABLE MySqlPersonAssetTable_copy(" +
                // declare the schema of the table 注意字段名用反引号转义
                "   trans_date STRING," +
                "   pay_code STRING," +
                "   trans_amt STRING," +
                "   trans_summary STRING," +
                "   trans_firm STRING," +
                "   trans_status STRING" +
                ") WITH(" +
                // declare the external system to connect to
                "   'connector'='jdbc'," +
                "   'url' = 'jdbc:mysql://127.0.0.1:3306/flink?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false&rewriteBatchedStatements=true'," +
                "   'table-name' = 'person_asset'," +
                "   'username' = 'root'," +
                "   'password' = '123456'" +
                ")");
        //2. using table descriptors create table with ElasticSearch connector

        TableResult tableResult = tEnv.executeSql("select * from MySqlPersonAssetTable");
        tableResult.print();
        // 查询
        // 1. query by Table API
        Table mySqlPersonAssetTable = tEnv.from("MySqlPersonAssetTable");
        Table tableQuery = mySqlPersonAssetTable.groupBy($("trans_firm"), $("trans_date"), $("pay_code"), $("trans_status"))
                .select($("trans_date"), $("pay_code"), $("trans_status"));

        // 2. query by SQL
        Table sqlQuery = tEnv.sqlQuery(
                "SELECT trans_date, pay_code, trans_status " +
                        "FROM MySqlPersonAssetTable " +
                        "GROUP BY trans_date, trans_firm, pay_code, trans_status ");

        // update (insert into ... select )

        tEnv.executeSql(
                "INSERT INTO MySqlPersonAssetTable_copy " +
                        "SELECT * " +
                        "FROM MySqlPersonAssetTable " );

        TableResult tableResult1 = tEnv.executeSql("select * from MySqlPersonAssetTable_copy");
        tableResult1.print();
    }

}
